package com.apes.framework.plugin.mq;

import com.apes.framework.plugin.esb.api.SimpleRequest;
import com.apes.framework.plugin.event.Listener;
import com.apes.framework.plugin.filereader.xml.ApesElement;
import com.apes.framework.plugin.mq.domain.MQueue;
import com.apes.framework.plugin.mq.domain.MQueueDeadLetter;
import com.apes.framework.plugin.mq.log.MQLog;
import com.apes.framework.plugin.mq.rocketmq.autoconfigure.ApesListenerContainerConfiguration;
import com.apes.framework.plugin.mq.rocketmq.autoconfigure.RocketMQMessageListener;
import com.apes.framework.plugin.mq.store.MQStoreService;
import com.apes.framework.util.*;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;

import static com.apes.framework.util.DateUtil.SHOW_DATETIME_MILLI_FORMAT;
import static com.apes.framework.util.DateUtil.format;

/**
 * 功能：Channel实现类
 *
 * @author xul
 * @create 2017-12-07 14:14
 */
@Service("messageChannel")
public class MessageChannel implements Channel {
    protected final Logger logger = LogManager.getLogger(MessageChannel.class);
    private static Map<String, Consumer> consumers = new HashMap();
    private static final int DEFAULT_MAX_RETRIES = 5;

    @Value("${rocketmq.consumer.client:false}")
    private boolean isConsumerClient;

    @Value("${rocketmq.consumer.consumeThreadMax:-1}")
    private int consumeThreadMax;

    @Value("${rocketmq.consumer.consumeTimeout:-1}")
    private int consumeTimeout;

    @Value("${rocketmq.tags:}")
    private String tags;

    @Value("${rocketmq.group.name:}")
    private String groupName;

    @Autowired
    private MQStoreService storeService;

    @Autowired
    private ApesListenerContainerConfiguration listenerContainerConfiguration;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private DistributedLock distributedLock;

    @Listener(topic = "event:mq.subscribe")
    public void subscribe(SimpleRequest request) {
        ApesElement nestedElement = (ApesElement) request.getData();
        MessageConsumer consumer = Tools.map(nestedElement.getAttributes(), MessageConsumer.class);
        consumer.setApesElement(nestedElement);
        addConsumer(consumer);
    }

    public void addConsumer(Consumer consumer) {
        String topic = consumer.getTopic();
        if (consumers.containsKey(topic)) {
            return;
        }
        RocketMQMessageListener annotation = new RocketMQMessageListener();
        annotation.setTopic(topic.replaceAll("\\.", "_").replaceAll("\\:", "_"));
        String consumeGroup = annotation.getTopic();
        if (StringUtils.hasText(this.groupName)) {
            consumeGroup = this.groupName + "%" + consumeGroup;
        }
        annotation.setConsumerGroup(consumeGroup);
        if (StringUtils.hasText(tags)) {
            annotation.setSelectorExpression(tags);
        }
        consumer.setRocketMQMessageListener(annotation);
        consumers.put(topic, consumer);
        if (isConsumerClient) {
            if (consumeThreadMax > 0) {
                annotation.setConsumeThreadMax(consumeThreadMax);
            }
            if (consumeTimeout > 0) {
                annotation.setConsumeTimeout(consumeTimeout);
            }
            listenerContainerConfiguration.registerContainer(annotation, new ApesRocketMQListener());
        }
    }

    @Listener(topic = "event:mq.unsubscribe")
    public void unsubscribe(SimpleRequest request) {
        String topic = request.get("topic");
        removeConsumer(topic);
    }

    public static void removeConsumer(String topic) {
        consumers.remove(topic);
    }

    @Override
    public void publish(MQueue message) {
        if (StringUtils.hasText(this.tags)) {
            message.setTags(this.tags);
        }
        message = storeService.save(message);
        String rocketMQTopic = getRocketMQTopic(message.getTopic(), this.tags);
        if (StringUtils.isEmpty(rocketMQTopic)) {
            return;
        }
        MessageHeaders headers = new MessageHeaders(MapUtil.mapper(RocketMQHeaders.KEYS, message.getId()));
        for (int i = 0; i < DEFAULT_MAX_RETRIES; i++) {
            try {
                rocketMQTemplate.convertAndSend(rocketMQTopic, message.getId(), headers);
                break;
            } catch (MessagingException e) {
                e.printStackTrace();
            }
        }
    }

    public String getRocketMQTopic(String messageTopic, String tags) {
        Map<String, Consumer> consumers = getConsumers();
        if (!consumers.containsKey(messageTopic)) {
            return null;
        }
        String rocketMQTopic = consumers.get(messageTopic).getRocketMQMessageListener().getTopic();
        if (StringUtils.hasText(tags)) {
            rocketMQTopic = rocketMQTopic + ":" + tags;
        }
        return rocketMQTopic;
    }

    class ApesRocketMQListener implements RocketMQListener<String> {

        @Override
        public void onMessage(String messageId) {
            distributedLock.lock(messageId, 120 * 1000, () -> handlingMessage(messageId), () -> {
                throw new RuntimeException("消息处理中，请稍等！");
            });
        }

        private void handlingMessage(String messageId) {
            MQLog log = null;
            try {
                MQueue message = storeService.findOne(messageId);
                if (message == null) {
                    return;
                }

                log = storeService.createLog(message);
                boolean isConsumed = false;
                Consumer consumer = null;
                if (consumers.containsKey(message.getTopic())) {
                    consumer = consumers.get(message.getTopic());
                    try {
                        MQEvent ropEvent = new MQEvent(this, RopUtil.getServiceRouter().getRopContext(), message);
                        isConsumed = consumer.consume(ropEvent);
                    } catch (Throwable e) {
                        e.printStackTrace();
                        message.setError(e.getMessage());
                        if (message.getTimes() == 0) {
                            sendExceptionMessage(message);
                        }
                        if (log != null) {
                            log.setError(Tools.getThrowable(e).getMessage() + "\n" + Tools.getAllStackTrace(e));
                        }
                    }
                } else {
                    message.setError("该消息没有消费者！");
                }

                message.incTimes();
                message.setEnded(format());
                if (isConsumed) {
                    storeService.delete(message);
                    if (consumer.isBackup()) {
                        storeService.backup(message);
                    }
                } else {
                    message.setState(1);
                    storeService.save(message);
                    if (message.getTimes() > 12) {
                        MQueueDeadLetter deadLetterQueue = new MQueueDeadLetter();
                        deadLetterQueue.setId(message.getId());
                        deadLetterQueue.setMsgId(message.getId());
                        String created = getDate(message.getCreated());
                        deadLetterQueue.setCreated(created);
                        storeService.save(deadLetterQueue);
                    }
                    throw new RuntimeException(message.getError()); /**异常是为了RocketMq消息不被消耗**/
                }
            } finally {
                if (log != null) {
                    storeService.updateLog(log);
                }
            }
        }

        private String getDate(String messageCreated) {
            try {
                return String.format("%s %s", format("yyyy-MM-dd"),
                        format(new SimpleDateFormat(SHOW_DATETIME_MILLI_FORMAT).parse(messageCreated), "HH:mm:ss.SSS"));
            } catch (ParseException e) {
                return DateUtil.format();
            }
        }

        /**
         * 发送异常消息
         */
        private void sendExceptionMessage(MQueue message) {
            EventUtil.sendAynEvent(this, "event:invoke.exception.intercept",
                    MapUtil.mapper("topic", message.getTopic(),
                            "description", message.getDescription(),
                            "type", "MQ",
                            "request", message.getBody(),
                            "response", message.getError()));
        }
    }

    public Map<String, Consumer> getConsumers() {
        return consumers;
    }

    @Listener(topic = "event:log.message.publish")
    public void logMessagePublish(SimpleRequest request) {
        MQUtil.publish("log.message", "日志消息", request.getData());
    }

    @com.apes.framework.plugin.mq.annotation.Consumer(topic = "log.message", backup = false, transaction = false)
    public void logMessageConsume(SimpleRequest request) {
        EventUtil.sendSynEvent(this, "event:log.message.consume", request.getData());
    }

    public void test(SimpleRequest request) {
        new ApesRocketMQListener().onMessage(request.get("messageId"));
    }
}
